package com.atguigu.app.dwd.log;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ProjectName: edu-realtime
 * @Package: com.atguigu.gmall.app.dwd.log
 * @ClassName DwdTrafficVideoPlayDetail
 * @Author jhh01
 * @Date 2022/9/3 13:59
 * @Description: ${测试代码}
 * @Version 1.0
 */
public class DwdTrafficVideoPlayDetail {
    public static void main(String[] args) throws Exception {
        //TODO 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        env.setStateBackend(new HashMapStateBackend());
         */
        //读取kafka 主题数据
        String topicName = "dwd_traffic_appvideo_log";
        String groupId = "dwd_traffic_video_play_detail";

         tableEnv.executeSql("create table video_play(\n" +
                "  `common` map<string,string> ,\n" +
                "  `appVideo` map<string,string> ,\n" +
                "  `ts` bigint \n" +
                ")" + KafkaUtil.getKafkaDDL(topicName, groupId));
        Table table = tableEnv.sqlQuery("select\n" +
                "  `common`['uid'] user_id,\n" +
                "  `appVideo`['play_sec'] play_sec,\n" +
                "  `appVideo`['position_sec'] position_sec,\n" +
                "  `appVideo`['video_id'] video_id,\n" +
                "  `ts` ts\n" +
                "from video_play");
        tableEnv.createTemporaryView("video_play", table);

        tableEnv.executeSql("create table dwd_video_play(\n" +
                "    `user_id` string ,\n" +
                "    `play_sec` string ,\n" +
                "    `position_sec` string ,\n" +
                "    `video_id` string ,\n" +
                "    `ts` bigint \n" +
                ")"+KafkaUtil.getKafkaSinkDDL("dwd_traffic_video_play_detail"));
        tableEnv.executeSql(" insert into dwd_video_play select * from video_play");
//        tableEnv.sqlQuery("select * from dwd_video_play").execute().print();


        /*Table video_infotable = tableEnv.fromDataStream(video_stream);
        tableEnv.sqlQuery("select * from video_infotable").execute().print();*/






    /*    env.execute(groupId);*/

    }
}
